// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_MEMORY_TABLE_SCANNER_HPP_
#define _XBASE_MEMORY_TABLE_SCANNER_HPP_

#include <queue>

#include "global.hpp"
#include "log_thread.hpp"

#include "kv_scanner.hpp"
#include "memory_table.hpp"

namespace xbase
{

	class memory_table_scanner
		: public kv_scanner
	{
	public:
		typedef boost::shared_ptr<memory_table_scanner> pointer;

		// 对多个内存表来说，也是应该保证，对Key相等的记录，要先从最新的内存表中取！
		struct impl_map_scan_state
		{
			memory_table::impl_map_pointer m_table;
			size_t m_seq_num; // 代表内存表新旧的序号，可看作是创建顺序，越大表示越新（类似于时间戳）
			bool m_need_check_lock;
			// 获取时，设置当时正在使用的内存表为True，其它
			// 为False；若在检查中发现不需要加锁，则将其置
			// 为False，使以后也不再检查
			
			row_block::pointer m_row_block; // 当前正在读的块，重要！它保证了这块内存不被修改和释放！
			size_t m_entry_index; // 下一次要读的块内索引偏移，可能超过当前块的块数，这时表示当前块已经读完了
			
			block_entry m_entry; // 每次next_entry后，就会
					     // 将结果保存到这里面，外
					     // 界从里面取的话先取这个
					     // 记录，再将其更新为下一
					     // 条

			impl_map_scan_state(memory_table::impl_map_pointer table,
					    const size_t &seq_num,
					    bool need_check_lock)
				: m_table(table)
				, m_seq_num(seq_num)
				, m_need_check_lock(need_check_lock)
				, m_row_block()
				, m_entry_index(0) {}
		};
		
		// 只比较要读的key，要求内部m_entry有效！
		struct greater_compare_for_impl_map_scan_state
		{
			inline
			bool operator()(const impl_map_scan_state &first,
					const impl_map_scan_state &second) const {
				key_equal_to key_eq;
				// second.entry.key < first.entry.key => first > second
				if(key_eq(*second.m_entry.m_key,
					  *first.m_entry.m_key)) // equal
				{
					// 作为时间戳来比较，新的（大的）是小的
					timestamp_less_than ts_lt;
					return ts_lt(second.m_seq_num,
						     first.m_seq_num);
				}
				else // not equal
				{
					key_less_than key_lt;
					return key_lt(*second.m_entry.m_key,
						      *first.m_entry.m_key);
				}
			}
		};

	public:
		memory_table_scanner(const kv_scan_context::pointer &p_context,
				     const memory_table::pointer &p_memtable)
			: kv_scanner(p_context)
			, m_memtable(p_memtable) {
			// set sequence_number
			// 用于在Key相等的时候，保证取到的是内存表的数据（最新），
			// 而不是SSTable的
			//
			// 用当前的日志序列号，可以保证比所有SSTable的序列号都大！
			m_sequence_number = global::the_log_manager->logger().get_sequence_number();
		}

		virtual
		~memory_table_scanner() {}

		// 如果队列为空，则为结束
		virtual
		bool eos() const {
			return m_queue.empty();
		}

		virtual
		const block_entry &get() const {
			assert(! m_queue.empty());
			return m_queue.top().m_entry;
		}

		// 在这里只需要检查行是否在范围内、列是否被包含即可
		virtual
		void next() {
			if(m_queue.empty())
			{
				LOG(DEBUG, "memory_table_scanner::next no table to scan, EOS");
				return;
			}

			impl_map_scan_state state = m_queue.top();
			m_queue.pop();

			const bool ok = next_entry(state);
			if(ok)	// 还有数据要读
			{
				m_queue.push(state);
			}
			// else drop it
			// 
			// 现在可以安全的抛弃掉了，因为只要保证当前记录的内存有效即可
			
			// else	// 若结束，则将他们保存到另一个列表中，
			// 	// 不能抛弃，否则可能会释放内存，使传出
			// 	// 去的指针失效。除非是第一次扫描，未传
			// 	// 出任何指针的情况下，可以抛弃。
			// {
			// 	m_eos_list.push_back(state);
			// }
		}

		//
		// 添加内存表的时候，要求指明该内存表的创建顺序
		// （seq_num），先创建的（老的）内存表，顺序号越小，当
		// 前内存表，顺序号最大（即最新）
		// 

		void add_history_table(const memory_table::impl_map_pointer &table,
				       const size_t &seq_num) {
			add_table(table, seq_num, false);
		}

		void add_working_table(const memory_table::impl_map_pointer &table,
				       const size_t &seq_num) {
			add_table(table, seq_num, true);
		}
		
	protected:
		void add_table(const memory_table::impl_map_pointer &table,
			       const size_t &seq_num,
			       bool is_working_table) {
			simple_buffer buffer;
			impl_map_scan_state state(table, seq_num, is_working_table);
			// 这里的buffer只在add_table中用到，专门用于lower_bound，用完后就不再用了。
			const key_wrapper *const least_key = scan_context_ptr()->get_least_key(buffer);
			// 用lower_bound来确定这个表内是否包含要扫描的数据，并将当前块、要读的块内索引设置好。
			// 它并不将数据从表中读取出来，所以还要用next_entry来真正的将需要的数据取出到state的m_entry字段
			const bool ok = lower_bound(state, least_key);
			buffer.consume(buffer.size());
			if(ok)	// 有要扫描的数据，进一步定位
			{
				const bool ok = next_entry(state); // 只有执行了next_entry后，设置了m_entry字段，才能放到队列中去
				if(ok)
				{
					m_queue.push(state);
				}
				//else 在这里，可以安全的把不需要的表抛弃掉，因为还没有传出属于它的指针。
			}
			//else, 在这里，可以安全的把不需要的表抛弃掉，因为还没有传出属于它的指针。
			//
			// lower_bound失败，说明这个表中没有要扫描的数据
		}

		// 如果当前块为空，则设为起始块；否则，设为当前块的下一行的块。同时设置块内索引为0。
		//
		// 结束返回false
		inline
		bool next_block_no_lock(impl_map_scan_state &impl_scan_state) const {
			memory_table::impl_map_type::iterator iter;
			if(! impl_scan_state.m_row_block) // 当前块为空
			{
				iter = impl_scan_state.m_table->begin();
			}
			else
			{
				iter = impl_scan_state.m_table->upper_bound(impl_scan_state.m_row_block->row_name());
			}
			
			if(iter == impl_scan_state.m_table->end()) // 读完了或是空表
			{
				return false;
			}
			
			impl_scan_state.m_row_block = iter->second.m_row_block;
			impl_scan_state.m_entry_index = 0;
			
			return true;
		}

		// 找到指定key属于哪个块，保存到 impl_map_scan_state 的
		// row_block中，并将下一行的块保存到 next_row_block 中，若没有
		// 下一行，则将其设置为空
		inline
		bool lower_bound_row_no_lock(impl_map_scan_state &scan_state,
					     row_block::pointer &next_row_block,
					     const key_wrapper *key) const {
			const memory_table::impl_map_type::iterator iter_end = scan_state.m_table->end();
			memory_table::impl_map_type::iterator iter = scan_state.m_table->lower_bound(key->row());
			if(iter_end == iter)
			{
				return false;
			}
			else
			{
				scan_state.m_row_block = iter->second.m_row_block;
				++iter;
				if(iter_end == iter)
				{
					next_row_block.reset(0); // 将其清空
				}
				else
				{
					next_row_block = iter->second.m_row_block;
				}
				return true;
			}
		}

		// 在这里只需要检查行是否在范围内、列是否被包含即可
		bool next_entry(impl_map_scan_state &scan_state) const;

		// 用来查找这个表中是否含有需要的Key，若有返回True，并
		// 设置扫描状态中的当前块为找到的块，并置块内索引为0；
		// 若没有返回False
		//
		// 经过lower_bound()的scan_state可以用于next_entry()
		bool lower_bound(impl_map_scan_state &scan_state,
				 const key_wrapper *key) const;
		
	private:
		// 当前实现只扫描正在使用的那个内存表，而不考虑正在紧缩的内存表
		memory_table::pointer m_memtable;

		std::priority_queue<impl_map_scan_state,
				    std::vector<impl_map_scan_state>,
				    greater_compare_for_impl_map_scan_state> m_queue;
	};
} // namespace xbase

#endif	// _XBASE_MEMORY_TABLE_SCANNER_HPP_
